package com.shujia.base;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.*;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;

import java.io.BufferedReader;
import java.io.FileReader;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;

/**
 *  hbase基础操作：
 *  1、建表
 *  2、插入一列数据
 *  3、获取一列数据
 *  4、插入一批多行数据
 *  5、获取一批数据
 *  6、删除表
 *  7、创建预分region表
 */

public class HBaseAPI {

    private Connection conn = null;
    private Admin admin = null;

    @Before
    public void init() {
        try {
            //使用HBase依赖种的类来创建与hbase的连接对象
            //HBaseConfiguration Deprecated
            //0.90.0版本后已经被弃用. 新版本种使用create()方法替代
//        HBaseConfiguration conf = new HBaseConfiguration();
            Configuration conf = HBaseConfiguration.create();
            //设置zookeeper的集群
            conf.set("hbase.zookeeper.quorum", "master,node1,node2");

            //通过配置文件，获取与hbase的连接对象
            conn = ConnectionFactory.createConnection(conf);
            System.out.println("成功获取数据库连接对象：" + conn);

            //获取数据库操作对象
            admin = conn.getAdmin();
            System.out.println("成功获取数据库操作对象：" + admin);

            System.out.println("---------------------------------------");

        } catch (Exception e) {
            e.printStackTrace();
        }

    }

    /**
     * 在hbase中创建一张表，表名叫做students
     */
    @Test
    public void createOneTable() {
        try {
            //将表名封装成TableName对象
            //TableName valueOf(String name)
            TableName name = TableName.valueOf("students");

            // 创建hbase表的必要条件是：表名和列簇
            //老版本中使用HTable类创建表对象
//            HTableDescriptor students = new HTableDescriptor(name);
            //2.0.0版本开始HTableDescriptor类已经被弃用，将会在3.0.0版本被移除
            //新版本推荐使用TableDescriptorBuilder类来获取一个表对象
            //1） 创建表对象
            TableDescriptorBuilder students = TableDescriptorBuilder.newBuilder(name);

            //2） 创建列簇对象
            //老版本中使用HColumnDescriptor类创建列簇对象
//            HColumnDescriptor info = new HColumnDescriptor("info");
            //新版中推荐使用ColumnFamilyDescriptorBuilder类中的of方法创建一个列簇对象
            ColumnFamilyDescriptor info = ColumnFamilyDescriptorBuilder.of("info");

            //3） 将列簇放到表中
//            students.addColumnFamily(info); // 老版本设置列簇
//            students.setColumnFamilies() // 新版本设置多个列簇，需要先将列簇对象封装在一个Collection
            students.setColumnFamily(info); // 新版本设置一个列簇

            //4） admin 数据库操作对象调用方法创建该表
            //createTable(TableDescriptor desc)
            admin.createTable(students.build());
            //最近java开发框架中，喜欢用xxxxBuilder()开始构建设置，设置完毕后调用一个build()方法创建对象

            System.out.println("成功创建：" + name);


        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    /**
     * 向hbase的students表中添加一列数据
     * 小虎，18，男，理科一班，女
     * put 'students','1001','info:name','小虎'
     * put 'students','1001','info:age','18'
     * put 'students','1001','info:gender','男'
     * put 'students','1001','info:clazz','理科一班'
     * put 'students','1001','info:like','女'
     */
    @Test
    public void putOneData() {
        try {
            //将表名封装成TableName对象
            //TableName valueOf(String name)
            TableName name = TableName.valueOf("students");

            // 获取目标表对象
            Table students = conn.getTable(name);

            //1) 构建Put对象, 一个Put仅代表一列【一个单元格】
            //Put(byte [] row) 传入行键的字节数组形式
            Put put = new Put(Bytes.toBytes("1001"));
            //进一步设置Put对象，完善要添加列的信息
            // 设置方式1：addColumn(byte [] family, byte [] qualifier, byte [] value)
//            put.addColumn(Bytes.toBytes("info"), Bytes.toBytes("name"), Bytes.toBytes("小虎"));
            //设置方式2：Put add(Cell cell)
            //将一列数据信息封装成Cell对象
            // Cell是一个接口，找到了一个实现类：KeyValue
            //KeyValue(final byte [] row, final byte [] family, final byte [] qualifier, final byte [] value)
            KeyValue cell = new KeyValue(Bytes.toBytes("1001"), Bytes.toBytes("info"), Bytes.toBytes("name"), Bytes.toBytes("小虎"));
            put.add(cell);


            //2) 调用方法添加一列数据
            students.put(put);

            System.out.println("成功添加一列数据！");
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    /**
     * 批量添加多列数据
     * 将student.txt数据添加到students表中
     */
    @Test
    public void putMoreData() {
        try {
            //将表名封装成TableName对象
            //TableName valueOf(String name)
            TableName name = TableName.valueOf("students");
            // 获取目标表对象
            Table students = conn.getTable(name);

            //使用java中的IO流操作读取student.txt文件
            BufferedReader br = new BufferedReader(new FileReader("data/student.txt"));
            String line = null;

            Put put = null;
            //定义一个列名的数组
            String[] colNameArray = {"id", "name", "age", "gender", "clazz"};

            //创建一个ArrayList集合保存每一个Put对象
            ArrayList<Put> puts = new ArrayList<>();

            while ((line = br.readLine()) != null) {
                //1500100001,施笑槐,22,女,文科六班
                //学号作为行键，其他当作普通单元格列存储
                String[] info = line.split(",");
                byte[] id = Bytes.toBytes(info[0]);
                //从数组第二个元素开始，每一个元素都要被封装成Put对象
                for (int i = 1; i < info.length; i++) {
                    put = new Put(id);
                    put.addColumn(Bytes.toBytes("info"), Bytes.toBytes(colNameArray[i]), Bytes.toBytes(info[i]));
                    puts.add(put);
                }
            }
            students.put(puts);
            System.out.println("student.txt表数据添加完毕！！");


        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    /**
     * 删除hbase中的一张表
     */
    @Test
    public void dropOneTable() {
        try {
            //将表名封装成TableName
            TableName t2 = TableName.valueOf("t2");

            //判断表是否存在，若存在，先禁用后删除
            if (admin.tableExists(t2)) {
                //禁用表
                admin.disableTable(t2);

                admin.deleteTable(t2);

                System.out.println("成功删除表：" + t2);
            } else {
                System.out.println(t2 + "表不存在！");
            }

        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    /**
     * 使用get方式获取数据，最多一行中多列
     * get 'students','1500100001'
     */
    @Test
    public void getOneData() {
        try {
            //将表名封装成TableName
            TableName tableName = TableName.valueOf("students");
            //获取表对象
            Table students = conn.getTable(tableName);

            //创建Get对象
            Get get = new Get(Bytes.toBytes("1500100044"));

            //使用表对象中的方法获取一条中的数据
            //返回值Result代表的是一条数据，包含这一行中的多列数据
            Result result = students.get(get);

            //分析Result，获取每一列数据 适合知道列名的情况下遍历
            //获取列数据的方式1：getValue(byte [] family, byte [] qualifier)
//            byte[] cf = Bytes.toBytes("info");
//            String id = Bytes.toString(result.getRow());
//            String name = Bytes.toString(result.getValue(cf, Bytes.toBytes("name")));
//            String age = Bytes.toString(result.getValue(cf, Bytes.toBytes("age")));
//            String gender = Bytes.toString(result.getValue(cf, Bytes.toBytes("gender")));
//            String clazz = Bytes.toString(result.getValue(cf, Bytes.toBytes("clazz")));
//            System.out.println("学号：" + id + ", 姓名：" + name + ", 年龄：" + age + ", 性别：" + gender + ", 班级：" + clazz);

            //获取列数据的方式2：List<Cell> listCells() //不知道列名的情况下遍历
            //获取所有列的单元格类型，遍历list集合获取每一个单元格列
            List<Cell> cells = result.listCells();
            String rk = Bytes.toString(result.getRow());
            StringBuilder sb = new StringBuilder();
            sb.append("rowKey：").append(rk).append(", ");
            for (Cell cell : cells) {
                //使用hbase中自带的工具类：CellUtil来获取一个单元格中的数据
//                String cf = Bytes.toString(CellUtil.cloneFamily(cell));
                String colName = Bytes.toString(CellUtil.cloneQualifier(cell));
                String colValue = Bytes.toString(CellUtil.cloneValue(cell));
                sb.append(colName).append("：").append(colValue).append(", ");
//                System.out.println("行键："+id+", 列簇"+cf+", 列名："+colName+", 列值："+colValue);
            }
            String info = sb.toString();
            info = info.substring(0,info.length()-2);
            System.out.println(info);

        } catch (Exception e) {
            e.printStackTrace();
        }
    }


    /**
     * 使用scan方法获取多行多列数据
     */
    @Test
    public void scanData(){
        try {
            //将表名封装成TableName
            TableName tableName = TableName.valueOf("students");
            //获取表对象
            Table students = conn.getTable(tableName);
            //创建一个Scan对象
            Scan scan = new Scan(); // 若使用无参构造方法创建Scan对象，将来获取的是整张表中的数据

            //设置开始行行键
//            scan.setStartRow(Bytes.toBytes("1500100665"));
//            //设置结束行行键
//            scan.setStopRow(Bytes.toBytes("1500100685"));
            //老版本传开始行，开始行会包含在结果中，传结束行，结束行不会包含在结果中

            //新版本中开始行和数据行都可以单独设置是否被包含在结果中
//            scan.withStartRow(Bytes.toBytes("1500100665"),false);
//            scan.withStopRow(Bytes.toBytes("1500100685"),false);

            scan.setLimit(5);


            //使用表对象中的getScanner方法获取批量行数据
            //ResultScanner 中包含了批量行数据
            ResultScanner resultScanner = students.getScanner(scan);
            //调用iterator方法获取包含所有行数据的迭代器
            for (Result result : resultScanner) {
                //分析一行中的每一列，就是上一个方法获取每一列的逻辑
                List<Cell> cells = result.listCells();
                String rk = Bytes.toString(result.getRow());
                StringBuilder sb = new StringBuilder();
                sb.append("rowKey：").append(rk).append(", ");
                for (Cell cell : cells) {
                    //使用hbase中自带的工具类：CellUtil来获取一个单元格中的数据
                    String colName = Bytes.toString(CellUtil.cloneQualifier(cell));
                    String colValue = Bytes.toString(CellUtil.cloneValue(cell));
                    sb.append(colName).append("：").append(colValue).append(", ");
                }
                String info = sb.toString();
                info = info.substring(0, info.length() - 2);
                System.out.println(info);
            }

        }catch (Exception e){
            e.printStackTrace();
        }
    }


    /**
     *  预分区表，设置切分键
     */
    @Test
    public void createPrepareTable(){
        try {
            TableName name = TableName.valueOf("tb_split");

            if(!admin.tableExists(name)){
                //创建表描述器对象
                TableDescriptorBuilder tbSplit = TableDescriptorBuilder.newBuilder(name);

                //创建列簇描述器对象
                ColumnFamilyDescriptor info = ColumnFamilyDescriptorBuilder.of("info");

                //将列簇加到表中
                tbSplit.setColumnFamily(info);

                byte[][] splitKeys = {
                        Bytes.toBytes("e"),
                        Bytes.toBytes("h"),
                        Bytes.toBytes("o"),
                        Bytes.toBytes("u")
                };

                //createTable(TableDescriptor desc, byte[][] splitKeys)
                //将分割键封装一个二维数组
                admin.createTable(tbSplit.build(), splitKeys);


                System.out.println("预分region表创建成功！");
            }else {
                System.out.println("该表已经存在！");
            }




        }catch (Exception e){
            e.printStackTrace();
        }
    }



    @After
    public void close() {
        try {
            if (admin != null) {
                admin.close();
            }

            if (conn != null) {
                conn.close();
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}
